package forkjoin;

import java.io.*;
import java.util.concurrent.*;

/**
 * Created by think on 2016/6/25.
 */
public class ProcessTxtTask extends RecursiveTask<String>{
    private File file;
    private int start;
    private int end;
    private static final int threshold = 10;

    public ProcessTxtTask(File file,int start, int end){
         this.file = file;
         this.start = start;
         this.end = end;
     }

    @Override
    protected String compute() {
        String result = "";
        if((end-start)<=threshold){   //处理
            return process(file,start,end);
        }else{                         //分割
            int middle = (start+end)/2;
            ProcessTxtTask ptt1 = new ProcessTxtTask(file,start,middle);
            ProcessTxtTask ptt2 = new ProcessTxtTask(file,middle,end);
            ptt1.fork();
            ptt2.fork();
            String s1 = ptt1.join();
            String s2 = ptt2.join();
            result = s1 + s2;
        }
        return result;
    }

    //处理文件中的字符串
    public String process(File file,int start,int end){
        String resultStr = "";
        //处理文件的start到end行
        try {
            RandomAccessFile randomAccessFile = new RandomAccessFile(file,"r");
            resultStr = randomAccessFile.readLine().split(":")[1];
            start++;
            while (resultStr != null && (start<=end)){
                resultStr += randomAccessFile.readLine().split(":")[1];
                start++;
            }
            randomAccessFile.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return resultStr;
    }

    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //生成总任务
        File file = new File("E:/key.txt");
        LineNumberReader  lnr = new LineNumberReader(new BufferedReader(new InputStreamReader(new FileInputStream(new File("E:/key.txt")))));
        //虽然跳过了9223372036854776000个字符，但它将返回实际上跳过的字符数160；long skip = lnr.skip(8) --> 跳过8个字符
        lnr.skip(Long.MAX_VALUE);
        //读取当前指针所在的行号
        int lineCount = lnr.getLineNumber();
        ProcessTxtTask processTxtTask = new ProcessTxtTask(file,0,lineCount);
        //向线程池提交任务
        Future<String> result = forkJoinPool.submit(processTxtTask);
        String resultStr = result.get();
        lnr.close();
        //监控线程池
        do {
            System.out.printf("Main: 当前线程池中活动的线程数Thread Count: %d\n", forkJoinPool.getActiveThreadCount());
            System.out.printf("Main: Thread Steal: %d\n", forkJoinPool.getStealCount());
            System.out.printf("Main: Parallelism: %d\n", forkJoinPool.getParallelism());
            try {
                TimeUnit.MILLISECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } while (!processTxtTask.isDone());
        forkJoinPool.shutdown();
        System.out.println("最终得到的字符串："+resultStr);
        long end = System.currentTimeMillis();
        System.out.println("消耗时间:"+(end-start)/1000);
    }
}
